package com.yanggu.flink.datastream_api.state.keyedstate

import cn.hutool.core.date.DateUtil
import com.yanggu.flink.datastream_api.pojo.Event
import com.yanggu.flink.datastream_api.source.ClickSource
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.util.Date

object ValueStateDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    env
      .addSource(new ClickSource)
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[Event] {
          override def extractTimestamp(element: Event, recordTimestamp: Long) = element.getTimestamp
        })
      )
      .keyBy(data => data.getName)
      //统计每个用户的 pv，隔一段时间（10s）输出一次结果
      .process(new PeriodicPvResult)
      .print()

    env.execute()

  }

}

//统计每个用户的 pv，隔一段时间（10s）输出一次结果
class PeriodicPvResult extends KeyedProcessFunction[String, Event, String] {

  private var pvState: ValueState[Long] = _

  private var timestampState: ValueState[Long] = _

  override def open(parameters: Configuration): Unit = {
    val pvStateDescriptor = new ValueStateDescriptor[Long]("pvState", classOf[Long])
    pvState = getRuntimeContext.getState(pvStateDescriptor)

    val timestampStateDescriptor = new ValueStateDescriptor[Long]("timestampState", classOf[Long])
    timestampState = getRuntimeContext.getState(timestampStateDescriptor)

  }

  override def processElement(value: Event, ctx: KeyedProcessFunction[String, Event, String]#Context, out: Collector[String]): Unit = {

    //更新pv次数, pv次数 + 1
    pvState.update(pvState.value() + 1L)

    val executeTimeStamp = timestampState.value()

    //如果为0表示没有注册定时器
    if (executeTimeStamp == 0L) {
      //增加一个10s后的时间事件定时器
      val ts = value.timestamp + 10 * 1000
      ctx.timerService().registerEventTimeTimer(ts)
      timestampState.update(ts)
    }
  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, Event, String]#OnTimerContext, out: Collector[String]): Unit = {
    val key = ctx.getCurrentKey
    val pvCount = pvState.value()
    out.collect(s"当前用户: $key, pv次数: $pvCount, 时间: ${DateUtil.formatDateTime(new Date(timestamp))}")

    //清空时间戳状态
    timestampState.clear()

  }

}
